/**
 * Copyright (c) 2013-2019 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;

import org.redisson.api.RFuture;
import org.redisson.api.RLock;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.RedisStrictCommand;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.pubsub.LockPubSub;

/**
 * 可重入公平锁。直接继承非公平锁
 * <p>
 * Distributed implementation of {@link java.util.concurrent.locks.Lock}
 * Implements reentrant lock.<br>
 * Lock will be removed automatically if client disconnects.
 * <p>
 * Implements a <b>fair</b> locking so it guarantees an acquire order by threads.
 *
 * @author Nikita Koksharov
 */
public class RedissonFairLock extends RedissonLock implements RLock {

    /**
     * 公平锁额外的属性。用于实现公平
     */
    // 时间差。强行让两个线程的过期时间存在一定的时间差，防止某个线程宕机。
    // 所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson会等待threadWaitTime毫秒后继续下一个线程。
    private final long threadWaitTime;
    private final CommandAsyncExecutor commandExecutor;
    // redisson_lock_queue:{lockName}
    private final String threadsQueueName;
    // redisson_lock_timeout:{lockName}
    private final String timeoutSetName;

    public RedissonFairLock(CommandAsyncExecutor commandExecutor, String name) {
        this(commandExecutor, name, 5000);
    }

    public RedissonFairLock(CommandAsyncExecutor commandExecutor, String name, long threadWaitTime) {
        super(commandExecutor, name);
        this.commandExecutor = commandExecutor;
        this.threadWaitTime = threadWaitTime;
        threadsQueueName = prefixName("redisson_lock_queue", name);
        timeoutSetName = prefixName("redisson_lock_timeout", name);
    }

    /**
     * 公平锁覆写的订阅通知机制。更改了entryName和channelName.
     * 不同于非公平锁，所有排队线程订阅同一个通道，一有锁释放的消息大家一起抢。
     * 公平锁，每个排队线程订阅单独的通道，锁释放的消息只会发给其中一个通道。
     *
     * @param threadId 当前线程Id
     * @return
     */
    @Override
    protected RFuture<RedissonLockEntry> subscribe(long threadId) {
        // param1  ConnectionId:lockName:ThreadId
        // param2  redisson_lock__channel:{lockName}:ConnectionId:ThreadId
        return pubSub.subscribe(getEntryName() + ":" + threadId,
                getChannelName() + ":" + getLockName(threadId));
    }

    @Override
    protected void unsubscribe(RFuture<RedissonLockEntry> future, long threadId) {
        pubSub.unsubscribe(future.getNow(), getEntryName() + ":" + threadId,
                getChannelName() + ":" + getLockName(threadId));
    }

    @Override
    protected RFuture<Void> acquireFailedAsync(long threadId) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_VOID,
                // get the existing timeout for the thread to remove
                "local queue = redis.call('lrange', KEYS[1], 0, -1);" +
                        // find the location in the queue where the thread is
                        "local i = 1;" +
                        "while i <= #queue and queue[i] ~= ARGV[1] do " +
                        "i = i + 1;" +
                        "end;" +
                        // go to the next index which will exist after the current thread is removed
                        "i = i + 1;" +
                        // decrement the timeout for the rest of the queue after the thread being removed
                        "while i <= #queue do " +
                        "redis.call('zincrby', KEYS[2], -tonumber(ARGV[2]), queue[i]);" +
                        "i = i + 1;" +
                        "end;" +
                        // remove the thread from the queue and timeouts set
                        "redis.call('zrem', KEYS[2], ARGV[1]);" +
                        "redis.call('lrem', KEYS[1], 0, ARGV[1]);",
                Arrays.<Object>asList(threadsQueueName, timeoutSetName),
                getLockName(threadId), threadWaitTime);
    }

    /**
     * 公平锁的redis结构:
     * 锁本身的结构和RedissonLock一样。
     * 一个名为 redisson_lock_queue:{lockName} 的列表。公平保存所有阻塞等待锁的 ConnectionId:ThreadId
     * 一个名为 redisson_lock_queue:{lockName} 的ZSET。内部保存v-s对。value是ConnectionId:ThreadId，score是他们的过期时间
     */

    /**
     * 覆写的 tryLockInnerAsync() 方法
     *
     * @param leaseTime 锁的过期时间。要么是用户指定的过期时间，要么是看门狗设置的过期时间
     * @param unit      时间单位
     * @param threadId  当前线程Id
     * @param command   ...
     * @param <T>
     * @return
     */
    @Override
    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        long currentTime = System.currentTimeMillis();

        // NULL转Boolean
        if (command == RedisCommands.EVAL_NULL_BOOLEAN) {
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                    // remove stale threads
                    "while true do " +
                            "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" +
                            "if firstThreadId2 == false then " +
                            "break;" +
                            "end;" +
                            "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" +
                            "if timeout <= tonumber(ARGV[3]) then " +
                            // remove the item from the queue and timeout set
                            // NOTE we do not alter any other timeout
                            "redis.call('zrem', KEYS[3], firstThreadId2);" +
                            "redis.call('lpop', KEYS[2]);" +
                            "else " +
                            "break;" +
                            "end;" +
                            "end;" +

                            "if (redis.call('exists', KEYS[1]) == 0) " +
                            "and ((redis.call('exists', KEYS[2]) == 0) " +
                            "or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then " +
                            "redis.call('lpop', KEYS[2]);" +
                            "redis.call('zrem', KEYS[3], ARGV[2]);" +

                            // decrease timeouts for all waiting in the queue
                            "local keys = redis.call('zrange', KEYS[3], 0, -1);" +
                            "for i = 1, #keys, 1 do " +
                            "redis.call('zincrby', KEYS[3], -tonumber(ARGV[4]), keys[i]);" +
                            "end;" +

                            "redis.call('hset', KEYS[1], ARGV[2], 1);" +
                            "redis.call('pexpire', KEYS[1], ARGV[1]);" +
                            "return nil;" +
                            "end;" +
                            "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                            "redis.call('hincrby', KEYS[1], ARGV[2], 1);" +
                            "redis.call('pexpire', KEYS[1], ARGV[1]);" +
                            "return nil;" +
                            "end;" +
                            "return 1;",
                    Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName),
                    internalLockLeaseTime, getLockName(threadId), currentTime, threadWaitTime);
        }

        // LONG转Boolean
        if (command == RedisCommands.EVAL_LONG) {
            // KEYS[1] = lockName
            // KEYS[2] = redisson_lock_queue:{lockName}
            // KEYS[3] = redisson_lock_timeout:{lockName}
            // ARGV[1] = internalLockLeaseTime
            // ARGV[2] = ConnectionId:ThreadId
            // ARGV[3] = threadWaitTime
            // ARGV[4] = currentTime
            return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                    // 开启一个while循环
                    "while true do " +
                        // 获取 redisson_lock_queue:{lockName} 列表的首个 ConnectionId:ThreadId
                        "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" +
                        // 如果 redisson_lock_queue:{lockName} 列表无元素，则退出while循环
                        "if firstThreadId2 == false then " +
                            "break;" +
                        "end;" +
                        // 获取firstThreadId2的过期时间
                        "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" +
                        // 如果已经到了过期时间，则从过期zset和队列中删除。否则不会继续判断之后的，直接退出while循环
                        "if timeout <= tonumber(ARGV[3]) then " +
                            "redis.call('zrem', KEYS[3], firstThreadId2);" +
                            "redis.call('lpop', KEYS[2]);" +
                        "else " +
                            "break;" +
                        "end;" +
                    "end;" +

                    // 检查此刻锁是否可以被获取: 锁已被释放，当前线程正好抢到或者当前线程是排队队列的首位
                    // (lockName不存在) AND (redisson_lock_queue:{lockName}不存在 OR redisson_lock_queue:{lockName}[0] == ConnectionId:ThreadId)
                    "if (redis.call('exists', KEYS[1]) == 0) " +
                        "and ((redis.call('exists', KEYS[2]) == 0) " +
                            "or (redis.call('lindex', KEYS[2], 0) == ARGV[2])) then " +

                        // 从zset和队列中删除(不用管存不存在，这里的删除的幂等操作)
                        "redis.call('lpop', KEYS[2]);" +
                        "redis.call('zrem', KEYS[3], ARGV[2]);" +

                        // 减少zset中排队线程的等待时间
                        "local keys = redis.call('zrange', KEYS[3], 0, -1);" +
                        "for i = 1, #keys, 1 do " +
                            "redis.call('zincrby', KEYS[3], -tonumber(ARGV[3]), keys[i]);" +
                        "end;" +

                        // 为当前线程创建锁，不需要处理zset和排队队列，因为无影响
                        "redis.call('hset', KEYS[1], ARGV[2], 1);" +
                        "redis.call('pexpire', KEYS[1], ARGV[1]);" +
                        "return nil;" +
                    "end;" +

                    // 如果锁已存在，则检查是否为重入锁
                    "if redis.call('hexists', KEYS[1], ARGV[2]) == 1 then " +
                        "redis.call('hincrby', KEYS[1], ARGV[2], 1);" +
                        "redis.call('pexpire', KEYS[1], ARGV[1]);" +
                        "return nil;" +
                    "end;" +

                    // 否则就是无法获取到锁，则检查当前线程是否早已处于排队队列中
                    "local timeout = redis.call('zscore', KEYS[3], ARGV[2]);" +
                    "if timeout ~= false then " +
                        // 如果已在排队中，则不操作redis，而是返回首次排队时设置的过期时间的剩余时间
                        "return timeout - tonumber(ARGV[3]) - tonumber(ARGV[4]);" +
                    "end;" +

                    // 获取排队队列最后一个(即上一个进入排队)的线程
                    "local lastThreadId = redis.call('lindex', KEYS[2], -1);" +
                    "local ttl;" +
                    // 如果排队队列非空，则设置 ConnectionId:ThreadId 的存活时间为上一个线程的剩余存活时间
                    "if lastThreadId ~= false and lastThreadId ~= ARGV[2] then " +
                        "ttl = tonumber(redis.call('zscore', KEYS[3], lastThreadId)) - tonumber(ARGV[4]);" +
                    // 如果排队队列为空，则设置 ConnectionId:ThreadId 的存活时间为锁的剩余存活时间
                    "else " +
                        "ttl = redis.call('pttl', KEYS[1]);" +
                    "end;" +
                    // ConnectionId:ThreadId 的过期时间就是 此刻+ttl+强制时间差
                    "local timeout = ttl + tonumber(ARGV[3]) + tonumber(ARGV[4]);" +
                    // 进zset和排队队列
                    "if redis.call('zadd', KEYS[3], timeout, ARGV[2]) == 1 then " +
                        "redis.call('rpush', KEYS[2], ARGV[2]);" +
                    "end;" +
                    // 返回剩余存活时间
                    "return ttl;",
                    Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName),
                    internalLockLeaseTime, getLockName(threadId), threadWaitTime, currentTime);
        }

        throw new IllegalArgumentException();
    }

    /**
     * 覆写的unlock()方法
     * @param threadId ThreadId
     * @return
     */
    @Override
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        // KEYS[1] = lockName
        // KEYS[2] = redisson_lock_queue:{lockName}
        // KEYS[3] = redisson_lock_timeout:{lockName}
        // KEYS[4] = redisson_lock__channel:{lockName}
        // ARGV[1] = 解锁通知 0L
        // ARGV[2] = 锁过期毫秒
        // ARGV[3] = ConnectionId:ThreadId
        // ARGV[4] = currentTime毫秒
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                // 和获取锁时一样，先删除已经排队队列中已经过期的
                "while true do " +
                    "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);" +
                    "if firstThreadId2 == false then " +
                        "break;" +
                    "end; " +
                    "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));" +
                    "if timeout <= tonumber(ARGV[4]) then " +
                        "redis.call('zrem', KEYS[3], firstThreadId2); " +
                        "redis.call('lpop', KEYS[2]); " +
                    "else " +
                        "break;" +
                    "end; " +
                "end;" +

                // 如果锁本身已经不存在了，那么直接给排队队列中的第一个线程订阅的通道发送锁释放的消息即可
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                    // 查看正在排队的下一个 ConnectionId:ThreadId
                    "local nextThreadId = redis.call('lindex', KEYS[2], 0); " +
                    "if nextThreadId ~= false then " +
                        // 向其订阅的通道 redisson_lock__channel:{lockName}:ConnectionId:ThreadId 发送锁释放的消息
                        "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
                    "end; " +
                    "return 1; " +
                "end;" +

                // 如果锁仍然存在，则需先判断持有锁的线程是否为当前 ConnectionId:ThreadId。不是也直接返回Nil
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                    "return nil;" +
                "end; " +

                // 否则是重入锁的解锁，只自减重入次数和更新过期时间。如果完全退出则删除锁并通知下一个人
                "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                "if (counter > 0) then " +
                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                    "return 0; " +
                "end; " +

                "redis.call('del', KEYS[1]); " +
                "local nextThreadId = redis.call('lindex', KEYS[2], 0); " +
                "if nextThreadId ~= false then " +
                    "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
                "end; " +
                "return 1; ",
                Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName, getChannelName()),
                LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId), System.currentTimeMillis());
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException();
    }

    @Override
    public RFuture<Boolean> deleteAsync() {
        return commandExecutor.writeAsync(getName(), RedisCommands.DEL_OBJECTS, getName(), threadsQueueName, timeoutSetName);
    }

    @Override
    public RFuture<Long> sizeInMemoryAsync() {
        List<Object> keys = Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName);
        return super.sizeInMemoryAsync(keys);
    }

    @Override
    public RFuture<Boolean> expireAsync(long timeToLive, TimeUnit timeUnit) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "redis.call('pexpire', KEYS[2], ARGV[1]); " +
                        "return redis.call('pexpire', KEYS[3], ARGV[1]); ",
                Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName),
                timeUnit.toMillis(timeToLive));
    }

    @Override
    public RFuture<Boolean> expireAtAsync(long timestamp) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "redis.call('pexpireat', KEYS[1], ARGV[1]); " +
                        "redis.call('pexpireat', KEYS[2], ARGV[1]); " +
                        "return redis.call('pexpireat', KEYS[3], ARGV[1]); ",
                Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName),
                timestamp);
    }

    @Override
    public RFuture<Boolean> clearExpireAsync() {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "redis.call('persist', KEYS[1]); " +
                        "redis.call('persist', KEYS[2]); " +
                        "return redis.call('persist', KEYS[3]); ",
                Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName));
    }


    @Override
    public RFuture<Boolean> forceUnlockAsync() {
        cancelExpirationRenewal(null);
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                // remove stale threads
                "while true do "
                        + "local firstThreadId2 = redis.call('lindex', KEYS[2], 0);"
                        + "if firstThreadId2 == false then "
                        + "break;"
                        + "end; "
                        + "local timeout = tonumber(redis.call('zscore', KEYS[3], firstThreadId2));"
                        + "if timeout <= tonumber(ARGV[2]) then "
                        + "redis.call('zrem', KEYS[3], firstThreadId2); "
                        + "redis.call('lpop', KEYS[2]); "
                        + "else "
                        + "break;"
                        + "end; "
                        + "end;"
                        +

                        "if (redis.call('del', KEYS[1]) == 1) then " +
                        "local nextThreadId = redis.call('lindex', KEYS[2], 0); " +
                        "if nextThreadId ~= false then " +
                        "redis.call('publish', KEYS[4] .. ':' .. nextThreadId, ARGV[1]); " +
                        "end; " +
                        "return 1; " +
                        "end; " +
                        "return 0;",
                Arrays.<Object>asList(getName(), threadsQueueName, timeoutSetName, getChannelName()),
                LockPubSub.UNLOCK_MESSAGE, System.currentTimeMillis());
    }

}